#include <thirdparty/fist-log.h>

#ifdef _MSC_VER
#include <errmsg.h>
#else
#include <mysql/errmsg.h>
#endif
#include <google/protobuf/descriptor.pb.h>

#include "db_info.h"
#include "db_mariadb.h"


CDBMariadb::CDBMariadb(IDBManager* db_manager)
        : m_db_manager(db_manager)
        , m_connected(false)
        , m_handle(nullptr)
        , m_need_stop(false)
        , m_last_connect_check(0)
{

}

CDBMariadb::~CDBMariadb()
{

}

int CDBMariadb::Init()
{
    mysql_library_init(0, nullptr, nullptr);
    TryConnect();
    m_exec_sql = std::thread(&CDBMariadb::ThreadRun, this);
    m_exec_sql.detach();
    return DB_OK;
}

int CDBMariadb::TryConnect()
{
    if (is_connected()) {
        return DB_OK;
    }
    const auto& connectKeys = m_db_manager->conf()->connectKeys;
    if (connectKeys.empty()) {
        LOG_ERROR("no db connect config. please check db.xml");
        return DB_FAIL;
    }
    const auto& con = connectKeys.front();
    m_handle = mysql_init(m_handle);
    MYSQL* newHandle = mysql_real_connect(m_handle, con.host.c_str(), con.username.c_str(), con.password.c_str(), "fist",
                                    con.port, nullptr, 0);
    if (newHandle == nullptr) {
        LOG_ERROR("connect fail. host:{} user:{} password:{} port:{}", con.host, con.username, con.password, con.port);
        return DB_CONNECT_FAIL;
    }
    m_handle = newHandle;
    set_connected(true);
    return DB_OK;
}

void CDBMariadb::Tick()
{
    CHECK_TIMEOUT(m_last_connect_check, 1000, TryConnect);
}

void CDBMariadb::Destroy()
{
    if (m_handle) {
        mysql_close(m_handle);
    }
    mysql_library_end();
    DBOperate* p = nullptr;
    while (!m_recycle_db_op.isEmpty()) {
        m_recycle_db_op.consume(p);
        if (p) {
            delete p ;
            p = nullptr;
        }
    }
}

DBOperate *CDBMariadb::NewDBOperator()
{
    while (m_recycle_db_op.isEmpty()) {
        auto pNew = new DBOperate;
        pNew->is_co = false;
        pNew->what = nullptr;
        m_recycle_db_op.produce(new DBOperate);
    }
    DBOperate* p = nullptr;
    m_recycle_db_op.consume(p);
    return p;
}

void CDBMariadb::set_connected(const bool &is_connected)
{
    this->m_connected = is_connected;
}

const bool &CDBMariadb::is_connected() const
{
    return m_connected;
}

#define CHECK_DB_GONE_AWAY(RET) ((RET) == CR_SERVER_GONE_ERROR || (RET) == CR_SERVER_LOST)

void CDBMariadb::ThreadRun()
{
    DBOperate* op = nullptr;
    int ret = 0;
    MYSQL_RES*  queryRes = nullptr;
    unsigned nField = 0;
    MYSQL_ROW rows;
    std::string fieldStr;
    while (!m_need_stop) {
        if (TryConnect() != DB_OK) {
            std::this_thread::sleep_for(std::chrono::seconds(1));
            continue;
        }
        op = nullptr;
        m_db_op.consume(op);
        if (op == nullptr) {
            LOG_ERROR("unknown sql!");
            continue;
        }
        ret = mysql_real_query(this->m_handle, op->sql.c_str(), op->sql.length());
        op->ret = ret;
        if (ret != 0) {
            if (CHECK_DB_GONE_AWAY(ret)) {
                set_connected(false);
            }
            LOG_FATAL("query error! sql:{}, err:{}", op->sql, mysql_error(m_handle));
            if (op->is_co) {
                op->lock.set();
            }
            m_recycle_db_op.produce(op);
            continue;
        }
        if (!op->is_co) {
            m_recycle_db_op.produce(op);
            continue;
        }
        // 下面是需要返回值的处理
        queryRes = mysql_store_result(m_handle);
        op->ret = mysql_errno(m_handle);
        if (mysql_errno(m_handle) != 0) {
            if (CHECK_DB_GONE_AWAY(mysql_errno(m_handle))) {
                set_connected(false);
            }
            LOG_FATAL("store result error! sql:{}, err:{}", op->sql, mysql_error(m_handle));
            op->lock.set();
            m_recycle_db_op.produce(op);
            continue;
        }
        if (queryRes == nullptr) {
            op->lock.set();
            m_recycle_db_op.produce(op);
            continue;
        }
        nField = mysql_num_fields(queryRes);
        auto temp = op->what;
        if (nField != op->what->GetDescriptor()->field_count()) {
            LOG_FATAL("field no match! db:{}, query:{}, sql:{}", nField, op->what->GetDescriptor()->field_count(), op->sql);
            op->lock.set();
            m_recycle_db_op.produce(op);
            continue;
        }

        // 获取数据
        while ((rows = mysql_fetch_row(queryRes)) != nullptr) {
            unsigned long *lengths = mysql_fetch_lengths(queryRes);
            auto r = temp->New();
            bool ok = true;
            for (int i = 0; i < nField; i++) {
                fieldStr.clear();
                if (rows[i] != nullptr) {
                    fieldStr.append(rows[i], lengths[i]);
                } else {
                    continue;
                }
                auto field = r->GetDescriptor()->field(i);
                switch (field->cpp_type()) {
                    case google::protobuf::FieldDescriptor::CPPTYPE_INT32:
                        r->GetReflection()->SetInt32(r, field, std::stoi(fieldStr));
                        break;
                    case google::protobuf::FieldDescriptor::CPPTYPE_INT64:
                        r->GetReflection()->SetInt64(r, field, std::stoll(fieldStr));
                        break;
                    case google::protobuf::FieldDescriptor::CPPTYPE_UINT32:
                        r->GetReflection()->SetUInt32(r, field, std::stoul(fieldStr));
                        break;
                    case google::protobuf::FieldDescriptor::CPPTYPE_UINT64:
                        r->GetReflection()->SetUInt64(r, field, std::stoull(fieldStr));
                        break;
                    case google::protobuf::FieldDescriptor::CPPTYPE_DOUBLE:
                        r->GetReflection()->SetDouble(r, field, std::stod(fieldStr));
                        break;
                    case google::protobuf::FieldDescriptor::CPPTYPE_FLOAT:
                        r->GetReflection()->SetFloat(r, field, std::stof(fieldStr));
                        break;
                    case google::protobuf::FieldDescriptor::CPPTYPE_STRING:
                        r->GetReflection()->SetString(r, field, fieldStr);
                        break;
                    default:
                        LOG_ERROR("no support filed type. type:{}", (int)field->cpp_type());
                        ok = false;
                        break;
                }
            }
            if (ok) {
                op->results->emplace_back(r);
            } else {
                break;
            }
        }
        mysql_free_result(queryRes);
        op->lock.set();
    }
}

void CombineFieldType(std::stringstream& buf, google::protobuf::Message* data)
{
    for (int i = 0; i < data->GetDescriptor()->field_count(); ++i) {
        auto field = data->GetDescriptor()->field(i);
        if (i != 0) {
            buf << ", ";
        }
        buf << field->name();
    }
}

bool ToStringStream(std::stringstream& buf, google::protobuf::Message* data,
                    const google::protobuf::FieldDescriptor* field)
{
    switch (field->cpp_type()) {
        case google::protobuf::FieldDescriptor::CPPTYPE_INT32:
            buf << data->GetReflection()->GetInt32(*data, field);
            break;
        case google::protobuf::FieldDescriptor::CPPTYPE_INT64:
            buf << data->GetReflection()->GetInt64(*data, field);
            break;
        case google::protobuf::FieldDescriptor::CPPTYPE_UINT32:
            buf << data->GetReflection()->GetUInt32(*data, field);
            break;
        case google::protobuf::FieldDescriptor::CPPTYPE_UINT64:
            buf << data->GetReflection()->GetUInt64(*data, field);
            break;
        case google::protobuf::FieldDescriptor::CPPTYPE_DOUBLE:
            buf << data->GetReflection()->GetDouble(*data, field);
            break;
        case google::protobuf::FieldDescriptor::CPPTYPE_FLOAT:
            buf << data->GetReflection()->GetFloat(*data, field);
            break;
        case google::protobuf::FieldDescriptor::CPPTYPE_STRING:
            buf << "\"" <<data->GetReflection()->GetString(*data, field) <<  "\"";
            break;
        default:
            LOG_ERROR("no support filed type. type:{}", (int)field->cpp_type());
            return false;
    }
    return true;
}

void CombineFieldValue(std::stringstream& buf, google::protobuf::Message* data)
{
    for (int i = 0; i < data->GetDescriptor()->field_count(); ++i) {
        auto field = data->GetDescriptor()->field(i);
        if (i != 0) {
            buf << ", ";
        }
        ToStringStream(buf, data, field);
    }
}

void GenerateInsertSql(std::stringstream& buf, google::protobuf::Message* data)
{
    buf << "insert into " << data->GetTypeName() << " (";
    CombineFieldType(buf, data);
    buf << ") values (";
    CombineFieldValue(buf, data);
    buf << ")";
}

int CDBMariadb::Insert(google::protobuf::Message *data)
{
    ResetSqlBuilder();
    GenerateInsertSql(m_sql_builder, data);
    DBOperate* p = NewDBOperator();
    p->sql = m_sql_builder.str();
    p->is_co = false;
    m_db_op.produce(p);
    return DB_OK;
}

cppcoro::task<int> CDBMariadb::CoInsert(google::protobuf::Message *data)
{
    ResetSqlBuilder();
    GenerateInsertSql(m_sql_builder, data);
    DBOperate* p = NewDBOperator();
    p->sql = m_sql_builder.str();
    p->is_co = true;
    p->lock.reset();
    p->ret = DB_OK;
    m_db_op.produce(p);
    co_await p->lock;
    co_return p->ret;
}

bool CombineDeleteFieldCondition(std::stringstream& buf, google::protobuf::Message* data)
{
    bool empty = true;
    for (int i = 0; i < data->GetDescriptor()->field_count(); ++i) {
        auto field = data->GetDescriptor()->field(i);
        if (!field->has_optional_keyword()) {
            continue;
        }
        for (int j = 0; j < google::protobuf::FieldOptions::GetDescriptor()->field_count(); ++j) {
            if (google::protobuf::FieldOptions::GetDescriptor()->field(j)->name() != DB_OPTION_KEY) {
                continue;
            }
            if (i != 0) {
                buf << ", ";
            }
            buf << field->name() << " = ";
            ToStringStream(buf, data, field);
            empty = false;
            break;
        }
    }
    return empty;
}

int CDBMariadb::Delete(google::protobuf::Message *data)
{
    ResetSqlBuilder();
    m_sql_builder << "delete " << data->GetTypeName() << " where ";
    if (CombineDeleteFieldCondition(m_sql_builder, data)) {
        LOG_INFO("not have key. ignore delete table:{}", data->GetTypeName());
        return DB_OK;
    }
    DBOperate* p = NewDBOperator();
    p->sql = m_sql_builder.str();
    p->is_co = false;
    m_db_op.produce(p);
    return DB_OK;
}

cppcoro::task<int> CDBMariadb::CoDelete(google::protobuf::Message *data)
{
    ResetSqlBuilder();
    m_sql_builder << "delete " << data->GetTypeName() << " where ";
    if (CombineDeleteFieldCondition(m_sql_builder, data)) {
        LOG_INFO("not have key. ignore delete table:{}", data->GetTypeName());
        co_return DB_OK;
    }
    DBOperate* p = NewDBOperator();
    p->sql = m_sql_builder.str();
    p->is_co = true;
    p->lock.reset();
    p->ret = DB_OK;
    m_db_op.produce(p);
    co_await p->lock;
    co_return p->ret;
}
const size_t MIN_DELETE_SQL_LENGTH = 8;
int CDBMariadb::Delete(const std::string &sql)
{
    if (sql.length() < MIN_DELETE_SQL_LENGTH) {
        LOG_ERROR("incorrect delete sql:{}", sql);
        return DB_FAIL;
    }
    DBOperate* p = NewDBOperator();
    p->sql = sql;
    p->is_co = false;
    m_db_op.produce(p);
    return DB_OK;
}

cppcoro::task<int> CDBMariadb::CoDelete(const std::string &sql)
{
    if (sql.length() < MIN_DELETE_SQL_LENGTH) {
        LOG_ERROR("incorrect delete sql:{}", sql);
        co_return DB_FAIL;
    }
    DBOperate* p = NewDBOperator();
    p->sql = sql;
    p->is_co = true;
    p->lock.reset();
    p->ret = DB_OK;
    m_db_op.produce(p);
    co_await p->lock;
    co_return p->ret;
}

bool CombineUpdateField(std::stringstream& buf, google::protobuf::Message* data)
{
    std::stringstream condition;
    bool haveKey = false;
    for (int i = 0; i < data->GetDescriptor()->field_count(); ++i) {
        auto field = data->GetDescriptor()->field(i);
        if (!field->has_optional_keyword()) {
            if (i != 0) {
                buf << ",";
            }
            buf << field->name() << " = ";
            ToStringStream(buf, data, field);
        } else {
            for (int j = 0; j < google::protobuf::FieldOptions::GetDescriptor()->field_count(); ++j) {
                if (google::protobuf::FieldOptions::GetDescriptor()->field(j)->name() != DB_OPTION_KEY) {
                    continue;
                }
                if (i != 0) {
                    condition << ", ";
                }
                condition << field->name() << " = ";
                if (ToStringStream(condition, data, field)) {
                    haveKey = true;
                }
                break;
            }
        }
    }
    if (!haveKey) {
        LOG_INFO("not have key. ignore update table:{}", data->GetTypeName());
        return false;
    }
    buf << " where " << condition.str();
    return true;
}

int CDBMariadb::Update(google::protobuf::Message *data)
{
    ResetSqlBuilder();
    m_sql_builder << "update " << data->GetTypeName() << "set ";
    if (!CombineUpdateField(m_sql_builder, data)) {
        return DB_OK;
    }
    DBOperate* p = NewDBOperator();
    p->sql = m_sql_builder.str();
    p->is_co = false;
    m_db_op.produce(p);
    return DB_OK;
}

cppcoro::task<int> CDBMariadb::CoUpdate(google::protobuf::Message *data)
{
    ResetSqlBuilder();
    m_sql_builder << "update " << data->GetTypeName() << "set ";
    if (!CombineUpdateField(m_sql_builder, data)) {
        co_return DB_OK;
    }
    DBOperate* p = NewDBOperator();
    p->sql = m_sql_builder.str();
    p->is_co = true;
    p->lock.reset();
    p->ret = DB_OK;
    m_db_op.produce(p);
    co_await p->lock;
    co_return p->ret;
}

const size_t MIN_UPDATE_SQL_LENGTH = 15;

int CDBMariadb::Update(const std::string &sql)
{
    if (sql.length() < MIN_UPDATE_SQL_LENGTH) {
        LOG_ERROR("incorrect update sql:{}", sql);
        return DB_FAIL;
    }
    DBOperate* p = NewDBOperator();
    p->sql = sql;
    p->is_co = false;
    m_db_op.produce(p);
    return DB_OK;
}

cppcoro::task<int> CDBMariadb::CoUpdate(const std::string &sql)
{
    DBOperate* p = NewDBOperator();
    p->sql = sql;
    p->is_co = true;
    p->lock.reset();
    p->ret = DB_OK;
    m_db_op.produce(p);
    co_await p->lock;
    co_return p->ret;
}

void GenerateReplaceSql(std::stringstream& buf, google::protobuf::Message* data)
{
    buf << "replace into " << data->GetTypeName() << " (";
    CombineFieldType(buf, data);
    buf << ") values (";
    CombineFieldValue(buf, data);
    buf << ")";
}

int CDBMariadb::Replace(google::protobuf::Message *data)
{
    ResetSqlBuilder();
    GenerateReplaceSql(m_sql_builder, data);
    DBOperate* p = NewDBOperator();
    p->sql = m_sql_builder.str();
    p->is_co = false;
    m_db_op.produce(p);
    return DB_OK;
}

cppcoro::task<int> CDBMariadb::CoReplace(google::protobuf::Message *data)
{
    ResetSqlBuilder();
    GenerateReplaceSql(m_sql_builder, data);
    DBOperate* p = NewDBOperator();
    p->sql = m_sql_builder.str();
    p->is_co = true;
    p->lock.reset();
    p->ret = DB_OK;
    m_db_op.produce(p);
    co_await p->lock;
    co_return p->ret;
}

const size_t MIN_REPLACE_SQL_LENGTH = 26;
int CDBMariadb::Replace(const std::string &sql)
{
    if (sql.length() < MIN_REPLACE_SQL_LENGTH) {
        LOG_ERROR("incorrect replace sql:{}", sql);
        return DB_FAIL;
    }
    DBOperate* p = NewDBOperator();
    p->sql = sql;
    p->is_co = false;
    m_db_op.produce(p);
    return DB_OK;
}

cppcoro::task<int> CDBMariadb::CoReplace(const std::string &sql)
{
    if (sql.length() < MIN_REPLACE_SQL_LENGTH) {
        LOG_ERROR("incorrect replace sql:{}", sql);
        co_return DB_FAIL;
    }
    DBOperate* p = NewDBOperator();
    p->sql = m_sql_builder.str();
    p->is_co = true;
    p->lock.reset();
    p->ret = DB_OK;
    m_db_op.produce(p);
    co_await p->lock;
    co_return p->ret;
}
/* select * from a */
const size_t MIN_SELECT_SQL_LENGTH = 15;

cppcoro::task<int> CDBMariadb::Select(const std::string &sql, google::protobuf::Message *what,
                                      std::list<google::protobuf::Message *> &results)
{
    if (sql.length() < MIN_SELECT_SQL_LENGTH) {
        LOG_ERROR("incorrect select sql:{}", sql);
        co_return DB_FAIL;
    }
    DBOperate* p = NewDBOperator();
    p->sql = sql;
    delete p->what;
    p->what = what;
    p->is_co = true;
    p->lock.reset();
    p->ret = DB_OK;
    p->results = &results;
    m_db_op.produce(p);
    co_await p->lock;
    co_return p->ret;
}

void CDBMariadb::ResetSqlBuilder()
{
    m_sql_builder.str("");
}
